package com.yanggu.redisson.component.consumer;


import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.dromara.hutool.core.array.ArrayUtil;
import org.dromara.hutool.core.collection.CollUtil;
import org.dromara.hutool.core.map.MapUtil;
import org.dromara.hutool.core.text.StrUtil;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.redisson.codec.TypedJsonJacksonCodec;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

/**
 * Redisson延迟队列注解Bean后置处理器。
 * <p>初始化所有的阻塞队列消费者</p>
 */
@Slf4j
@Component
public class RedissonDelayedQueueListenerBeanPostProcessor implements InitializingBean, DisposableBean {

    @Value("${delayed.queue.core-pool-size:10}")
    private Integer corePoolSize;

    @Value("${delayed.queue.max-pool-size:20}")
    private Integer maxPoolSize;

    @Value("${delayed.queue.keep-alive-seconds:60}")
    private Integer keepAliveSeconds;

    @Value("${delayed.queue.queue-capacity:100}")
    private Integer queueCapacity;

    @Value("${delayed.queue.thread-name-prefix:DelayedQueue-}")
    private String threadNamePrefix;

    private Map<String, ThreadPoolTaskExecutor> executorMap;

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private Environment environment;

    @Autowired
    private ObjectMapper objectMapper;

    @Override
    public void afterPropertiesSet() {
        log.info("Initializing redisson delayed queue consumer");
        //扫描Component注解的Bean
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(Component.class);
        if (MapUtil.isEmpty(beans)) {
            return;
        }
        //获取所有Bean，方法包含RedissonDelayedQueueListener的方法
        List<ConsumerParam> list = getConsumerParams(beans);
        if (CollUtil.isEmpty(list)) {
            return;
        }
        handlerListener(list);
        log.info("Started redisson delayed queue consumer");
    }

    @Override
    public void destroy() {
        log.info("Shutting down redisson delayed queue consumer");
        if (MapUtil.isEmpty(executorMap)) {
            return;
        }
        //关闭线程池
        executorMap.values().forEach(threadPoolTaskExecutor -> {
            log.info("Shutting down redisson delayed queue consumer thread pool, name: {}", threadPoolTaskExecutor.getThreadNamePrefix());
            threadPoolTaskExecutor.destroy();
        });
    }

    private List<ConsumerParam> getConsumerParams(Map<String, Object> beans) {
        List<ConsumerParam> list = new ArrayList<>();
        for (Object bean : beans.values()) {
            Method[] declaredMethods = bean.getClass().getDeclaredMethods();
            if (ArrayUtil.isEmpty(declaredMethods)) {
                continue;
            }
            for (Method method : declaredMethods) {
                if (!method.isAnnotationPresent(RedissonDelayedQueueListener.class)) {
                    continue;
                }
                RedissonDelayedQueueListener annotation = method.getAnnotation(RedissonDelayedQueueListener.class);
                String queueName = annotation.value();
                if (StrUtil.isBlank(queueName)) {
                    continue;
                }
                //解析队列名称，支持spring配置
                queueName = environment.resolveRequiredPlaceholders(queueName);
                if (StrUtil.isBlank(queueName)) {
                    continue;
                }
                TypeReference<Object> typeReference = new TypeReference<>() {
                    @Override
                    public Type getType() {
                        //获取方法第一个参数的完整泛型类型
                        return method.getParameters()[0].getParameterizedType();
                    }
                };
                //使用自定义的Codec，为jackson方式。可以支持复杂泛型
                TypedJsonJacksonCodec codec = new TypedJsonJacksonCodec(typeReference);
                RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(queueName, codec);
                Consumer<Object> consumer = getConsumer(bean, method, queueName);
                ConsumerParam consumerParam = new ConsumerParam();
                consumerParam.setBlockingQueue(blockingQueue);
                consumerParam.setConsumer(consumer);
                consumerParam.setAnnotation(annotation);
                consumerParam.setMethod(method);
                list.add(consumerParam);
            }
        }
        return list;
    }

    private void handlerListener(List<ConsumerParam> list) {
        executorMap = new HashMap<>();
        Map<String, ThreadPoolTaskExecutor> beanMap = applicationContext.getBeansOfType(ThreadPoolTaskExecutor.class);
        //循环处理队列
        list.forEach(consumerParam -> {
            String name = consumerParam.getBlockingQueue().getName();
            String executorBeanName = consumerParam.getAnnotation().executorBeanName();
            ThreadPoolTaskExecutor taskExecutor = null;
            //如果指定了线程池名称，则使用指定的线程池
            if (StrUtil.isNotBlank(executorBeanName)) {
                taskExecutor = beanMap.get(executorBeanName);
            }
            //如果没有指定线程池，则使用队列名称作为线程池名称
            if (taskExecutor == null) {
                taskExecutor = beanMap.get(name);
            }
            //如果没有找到线程池，则使用默认参数初始化线程池
            if (taskExecutor == null) {
                //初始化线程池
                taskExecutor = initExecutor(name);
            }
            executorMap.put(name, taskExecutor);
            log.info("Starting redisson delayed queue listener for queue: {}, method: {}", name, consumerParam.getMethod());
            processNextMessage(consumerParam.getBlockingQueue(), consumerParam.getConsumer());
        });
    }

    private Consumer<Object> getConsumer(Object bean, Method method, String queueName) {
        return (Object data) -> {
            String jsonString = getJsonString(data);
            try {
                log.info("Processing redisson delayed queue message, queue: {}, message: {}", queueName, jsonString);
                method.invoke(bean, data);
            } catch (Exception e) {
                log.error("Error processing redisson delayed queue message, queue: {}, message: {}", queueName, jsonString, e);
            }
        };
    }

    @SneakyThrows
    private String getJsonString(Object data) {
        return objectMapper.writeValueAsString(data);
    }

    private ThreadPoolTaskExecutor initExecutor(String name) {
        String prefix = threadNamePrefix + name + "-";
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //使用配置的核心线程数
        taskExecutor.setCorePoolSize(corePoolSize);
        //使用配置的最大线程数
        taskExecutor.setMaxPoolSize(maxPoolSize);
        //使用配置的存活时间
        taskExecutor.setKeepAliveSeconds(keepAliveSeconds);
        //使用配置的队列容量
        taskExecutor.setQueueCapacity(queueCapacity);
        //使用配置的线程名前缀
        taskExecutor.setThreadNamePrefix(threadNamePrefix + name + "-");
        //拒绝策略
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //预启动所有核心线程
        taskExecutor.setPrestartAllCoreThreads(true);
        taskExecutor.initialize();
        log.info("Initialized redisson delayed queue consumer executor with core pool size: {}, max pool size: {}, keep alive seconds: {}, queue capacity: {}, thread name prefix: {}",
                corePoolSize, maxPoolSize, keepAliveSeconds, queueCapacity, prefix);
        return taskExecutor;
    }

    private void processNextMessage(RBlockingQueue<Object> blockingQueue, Consumer<Object> consumer) {
        blockingQueue.takeAsync()
                .thenAcceptAsync(consumer, executorMap.get(blockingQueue.getName()))
                .exceptionally(throwable -> {
                    log.error("Error processing redisson delayed queue message", throwable);
                    //发生异常了sleep一会儿。可能是redis连不上了
                    retryWithDelay();
                    return null;
                })
                .thenRunAsync(() -> processNextMessage(blockingQueue, consumer), executorMap.get(blockingQueue.getName()));
    }

    private void retryWithDelay() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException ignored) {
        }
    }

    @Data
    static class ConsumerParam {

        private RBlockingQueue<Object> blockingQueue;

        private Consumer<Object> consumer;

        private RedissonDelayedQueueListener annotation;

        private Method method;

    }

}